package com.dark.trident;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import storm.trident.operation.BaseFunction;
import storm.trident.operation.TridentCollector;
import storm.trident.tuple.TridentTuple;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Created by tengxue on 16-6-30.
 */
public class CityAssignment  extends  BaseFunction{

    private static final long serialVersionUID=1L;

    private static final Logger LOG= LoggerFactory.getLogger(CityAssignment.class);

    private static Map<String,double[]> CITIES=new HashMap<String,double[]>();
    static {
        double[] phl={39.875365,-75.249524};
        CITIES.put("PHL",phl);
        double[] nyc={40.71448,-74.00598};
        CITIES.put("NYC",nyc);
        double[] sf={-31.4250142,-62.0841809};
        CITIES.put("SF",sf);
        double[] la={-34.05374,-118.24307};
        CITIES.put("LA",la);

    }

    public void execute(TridentTuple tuple, TridentCollector collector) {
        DiagnosisEvent diagnosis=(DiagnosisEvent)tuple.get(0);
        double leastDistance=Double.MAX_VALUE;
        String closestCity="NONE";

        //计算最近的城市
        for (Map.Entry<String,double[]> city:CITIES.entrySet()){
            double R=6371;// km
            double x=(city.getValue()[0]-diagnosis.getLng())*
                    Math.cos((city.getValue()[0]+diagnosis.getLng())/2);
            double y=(city.getValue()[1]-diagnosis.getLat());
            double d=Math.sqrt(x*x+y*y)*R;
            if (d<leastDistance){
                leastDistance=d;
                closestCity=city.getKey();
            }
        }

        //emit the value
        List<Object> values=new ArrayList<Object>();
        values.add(closestCity);
        LOG.info("Closest city : "+diagnosis);
        collector.emit(values);

    }
}
